package ckR

//
//import (
//	"fmt"
//	"github.com/google/uuid"
//	"db2s/global"
//	mq "db2s/topic-mq"
//	"math/rand"
//	"strings"
//	"sync"
//	"sync/atomic"
//	"time"
//)
//
//// rowsMissSegmentQueryListeners rowsMissSegmentQuery  监听执行器
//func rowsMissSegmentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName       = string(message.Body)
//		plan            = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		logseq          = message.Properties["logSeq"].(int64)
//		topic           = RowsMissMd5
//		dTopic          = rowsMissSegmentQuery
//		event           = "[rowsMissQuerySqlListeners]"
//		wg              sync.WaitGroup
//		fatherSonTags   = make(map[string][]string)
//		done            = make(chan struct{})
//		missDataPointer chan any
//		deleteMapInfo   = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	plan.subTaskBeginTime = time.Now().UnixNano()
//	if GlobalPConfigs.result.Teletypewriter == "bar" {
//		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
//		global.Wlog.Info(fmt.Sprintf("(%d) %s Add task Bar Run object %v successfully", logseq, event, tableName))
//	}
//	wg.Add(2)
//	//init status
//	go expensiveOperation(&wg, message, done, 1, 1)
//	go mqMissTableStatusInterruptMonitor(done, &wg, message, logseq)
//
//	plan.subTaskBeginTime = time.Now().UnixNano()
//	//处理wheresql
//	if v, ok := GlobalPConfigs.schema.WhereSql[fmt.Sprintf("%s%s%s", plan.schema, global.SchemaTableSplist, plan.table)]; ok && v != "" {
//		//global.Wlog.Info(fmt.Sprintf("(%d) %s Currently there is a [wheresql] mode, begin sending topic %v", logseq, event, topic))
//		//sqlWhere = make(chan any, 1)
//		//sqlWhere <- map[string]string{"whereSql": v}
//	} else {
//		wg.Add(2)
//		sourceDataPointer, destDataPointer, err := plan.rowsMissSegmentQuery(&wg, logseq, event)
//		if err != nil {
//			plan.errorAction(message, logseq, err)
//			return mq.ReconsumeLater
//		}
//		missDataPointer = plan.blockingReadPipeData(&wg, sourceDataPointer, destDataPointer)
//	}
//	uid := uuid.New()
//	if !chanMonitorSendMsg(missDataPointer,
//		getCheckMod(), topic, tableName, sendMsg, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//			"taskObjectNeglect": true, "label": map[string]any{uid.String(): []string{}},
//			"sendMsgKey": "chunkData", "subTask": plan.subTaskInfo, "topic": string(topic),
//			"outFay": GlobalPConfigs.result.Teletypewriter}, fatherSonTags, func() {
//		}) {
//		return mq.ReconsumeLater
//	}
//	if atomic.LoadInt64(plan.errEnd) > -1 {
//		plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
//	}
//	atomic.SwapInt64(plan.mqProductCustomerMonitor.mqStartTiming, 1)
//	wg.Wait()
//	return mq.ConsumeSuccess
//}
//
//// rowsMissMd5Listeners rowsMissMd5  监听执行器
//func rowsMissMd5Listeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName                      = string(message.Body)
//		logseq                         = message.Properties["logSeq"].(int64)
//		plan                           = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic                          = RowsMissArrcmp
//		dTopic                         = RowsMissMd5
//		event                          = "[rowsMissMd5Listeners]"
//		srcDataPointer, dstDataPoniter []*string
//		taskObject                     = message.Properties["taskObject"]
//		taskObjectNeglect              = message.Properties["taskObjectNeglect"].(bool)
//		deleteMapInfo                  = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if !plan.sendDatafixMqMsgStatus() {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ReconsumeLater
//	}
//	a := message.Properties["chunkData"].(map[string]any)
//	if src, ok := a["src"]; ok {
//		srcDataPointer = src.([]*string)
//	}
//	if dst, ok := a["dst"]; ok {
//		dstDataPoniter = dst.([]*string)
//	}
//	stt1, dtt1, abs := plan.rowsMissMd5Rows(srcDataPointer, dstDataPoniter)
//	if abs {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ConsumeSuccess
//	}
//	plan.modifyDatafixMqMsgStatus()
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"sourceD": stt1, "destD": dtt1, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsMissArrcmpListeners rowsMissArrcmp 监听执行器
//func rowsMissArrcmpListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName         = string(message.Body)
//		logseq            = message.Properties["logSeq"].(int64)
//		plan              = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic             = RowsMissAbnormalDataDiff
//		dTopic            = RowsMissArrcmp
//		event             = "[rowsMissArrcmpListeners]"
//		taskObject        = message.Properties["taskObject"]
//		taskObjectNeglect = message.Properties["taskObjectNeglect"].(bool)
//		deleteMapInfo     = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if !plan.sendDatafixMqMsgStatus() {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ReconsumeLater
//	}
//	stt, dtt := message.Properties["sourceD"].(*string), message.Properties["destD"].(*string)
//	dataDisPos := plan.rowsMissArrcmp(stt, dtt, logseq)
//	if len(dataDisPos) == 0 {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": taskObject, "taskObjectNeglect": taskObjectNeglect, "label": message.Properties["label"],
//		"dataDisPos": dataDisPos, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsMissAbnormalDataDiffListeners rowsMissAbnormalDataDiff 监听执行器
//func rowsMissAbnormalDataDiffListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		dTopic        = RowsMissAbnormalDataDiff
//		event         = "[rowsMissAbnormalDataDiffListeners]"
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if !plan.sendDatafixMqMsgStatus() {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ReconsumeLater
//	}
//	dataDisPos := message.Properties["dataDisPos"].(map[string][]*string)
//	if err := plan.rowsMissAbnormalDataPersistence(dataDisPos); err != nil {
//		plan.errorAction(message, logseq, err)
//		//atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ReconsumeLater
//	}
//	atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//	return mq.ConsumeSuccess
//}
//
//// rowsMissMd5ValueUniqListeners rowsMissMd5ValueUniq 监听执行器
//func rowsMissMd5ValueUniqListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		logseq        = rand.NewSource(time.Now().UnixNano()).Int63()
//		topic         = RowsMissDataValueRead
//		dTopic        = RowsMissMd5ValueUniq
//		event         = "[rowsMissMd5ValueUniqListeners]"
//		wg            sync.WaitGroup
//		done          = make(chan struct{})
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	wg.Add(2)
//	go expensiveOperation(&wg, message, done, 1, 1)
//	go mqTableStatusMonitor(done, &wg, plan, logseq)
//	if e := atomic.LoadInt64(plan.errEnd); e < 0 {
//		return mq.ReconsumeLater
//	}
//	plan.mqProductCustomerMonitor.ProductMessage = make(chan map[string][]string)
//	chunkData := plan.rowsMissMd5ValueUniq()
//	//send message
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"md5UniqueConstraint": chunkData, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	atomic.SwapInt64(plan.mqProductCustomerMonitor.mqStartTiming, 0)
//	wg.Wait()
//	return mq.ConsumeSuccess
//}
//
//// rowsMissDataValueReadListeners rowsMissDataValueRead  监听执行器
//func rowsMissDataValueReadListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		logseq        = rand.NewSource(time.Now().UnixNano()).Int63()
//		topic         = RowsMissDataGeneralSql
//		dTopic        = RowsMissDataValueRead
//		event         = "[rowsMissDataValueReadListeners]"
//		fatherSonTags = make(map[string][]string)
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if message.Properties["md5UniqueConstraint"] == nil {
//		return mq.ReconsumeLater
//	}
//	md5uniq := message.Properties["md5UniqueConstraint"].(map[string]int)
//	abnormalDataPointer := plan.rowsMissDataValueRead(md5uniq)
//	uid := uuid.New()
//	if !chanMonitorSendMsg(abnormalDataPointer,
//		getCheckMod(), topic, tableName, sendMsg, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//			"taskObjectNeglect": true, "label": map[string]any{uid.String(): []string{}},
//			"sendMsgKey": "chunkData", "subTask": plan.subTaskInfo, "topic": string(topic),
//			"outFay": GlobalPConfigs.result.Teletypewriter}, fatherSonTags, func() {
//		}) {
//		return mq.ReconsumeLater
//	}
//	plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
//	atomic.SwapInt64(plan.mqProductCustomerMonitor.mqStartTiming, 1)
//	return mq.ConsumeSuccess
//}
//
//// rowsMissDataGeneralSqlListeners rowsMissDataGeneralSql  监听执行器
//func rowsMissDataGeneralSqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		topic         = RowsMissFixSqlExec
//		dTopic        = RowsMissDataGeneralSql
//		event         = "[rowsMissDataGeneralSqlListeners]"
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if e := atomic.LoadInt64(plan.errEnd); e < 0 {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//		return mq.ReconsumeLater
//	}
//	chunkData := message.Properties["chunkData"]
//	_, _, dataFixSql, err := plan.rowsMissDataGeneralSql(chunkData)
//	if err != nil {
//		plan.errorAction(message, logseq, err)
//		return mq.ReconsumeLater
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"label":      message.Properties["label"],
//		"dataFixSql": dataFixSql, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsMissFixSqlExecListeners rowsMissFixSqlExec  监听执行器
//func rowsMissFixSqlExecListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(map[string]*SchedulePlan2)[tableName]
//		dTopic        = RowsMissFixSqlExec
//		event         = "[rowsMissFixSqlExecListeners]"
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	if !plan.sendDatafixMqMsgStatus() {
//		return mq.ReconsumeLater
//	}
//	if !strings.EqualFold(GlobalPConfigs.repair.DataFix, "no") {
//		dataFixSql := message.Properties["dataFixSql"].(map[string][]*string)
//		if err := plan.rowsMissDataFixDiff(dataFixSql, logseq); err != nil {
//			plan.errorAction(message, logseq, err)
//			return mq.ReconsumeLater
//		}
//	}
//	atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
//	return mq.ConsumeSuccess
//}
